/*
 * Copyright (c) 2012-2014 The original author or authors
 * ------------------------------------------------------
 * All rights reserved. This program and the accompanying materials
 * are made available under the terms of the Eclipse Public License v1.0
 * and Apache License v2.0 which accompanies this distribution.
 *
 * The Eclipse Public License is available at
 * http://www.eclipse.org/legal/epl-v10.html
 *
 * The Apache License v2.0 is available at
 * http://www.opensource.org/licenses/apache2.0.php
 *
 * You may elect to redistribute this code under either of these licenses.
 */
package org.dna.mqtt.moquette.processor.impl;

import org.dna.mqtt.moquette.messaging.spi.impl.ValueEvent;
import org.dna.mqtt.moquette.messaging.spi.impl.events.OutputMessagingEvent;
import org.dna.mqtt.moquette.messaging.spi.impl.events.PublishEvent;
import org.dna.mqtt.moquette.proto.messages.AbstractMessage;
import org.dna.mqtt.moquette.proto.messages.PubCompMessage;
import org.dna.mqtt.moquette.proto.messages.PubRelMessage;
import org.dna.mqtt.moquette.server.Constants;
import org.dna.mqtt.moquette.server.ServerChannel;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * @author puran
 */
public class PubRelProcessor extends AbstractProtocolProcessor {
	private static final Logger LOG = LoggerFactory.getLogger(PubRelProcessor.class);
	
	public PubRelProcessor(ServerChannel session, PubRelMessage message) {
		super(session, message);
	}

	/**
     * Second phase of a publish QoS2 protocol, sent by publisher to the broker. Search the stored message and publish
     * to all interested subscribers.
     */
	public void processEvent(ValueEvent valueEvent) {
		ServerChannel session = getSession();
		PubRelMessage message = (PubRelMessage)getMessage();
		
		String clientID = (String) session.getAttribute(Constants.ATTR_CLIENTID);
        int messageID = message.getMessageID();
        
        MqttProtocolProcessor baseProcessor = getBaseProcessor();
        
        LOG.debug("PUB --PUBREL--> SRV processPubRel invoked for clientID {} ad messageID {}", clientID, messageID);
        String publishKey = String.format("%s%d", clientID, messageID);
        PublishEvent evt = baseProcessor.retrieveQoS2Message(publishKey);

        final String topic = evt.getTopic();
        final AbstractMessage.QOSType qos = evt.getQos();
        
        baseProcessor.publish2Subscribers(topic, qos, evt.getMessage(), evt.isRetain(), evt.getMessageID());
        baseProcessor.removeQoS2Message(publishKey);

        if (evt.isRetain()) {
        	baseProcessor.storeRetained(topic, evt.getMessage(), qos);
        }

        LOG.debug("PUB <--PUBCOMP-- SRV sendPubComp invoked for clientID {} ad messageID {}", clientID, messageID);
        PubCompMessage pubCompMessage = new PubCompMessage();
        pubCompMessage.setMessageID(messageID);
        
        ServerChannel clientSession = baseProcessor.getClientSession(clientID);
        
        baseProcessor.disruptorPublish(new OutputMessagingEvent(clientSession, pubCompMessage));
	}
}
